﻿using DH.Common;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace DH.MQ
{
    public class RMQProvider
    {
        private XMLConfig cs;
        private readonly object _loadLock = new object();

        private RabbitMQContext context = new RabbitMQContext();

        public BaseMessage basemessage;
        public string ListenQueueName;

        public Func<BaseMessage, object> fun = null;
        public Action<BaseMessage> act = null;

       /// <summary>
       /// 
       /// </summary>
       /// <param name="message">消息内容</param>
       /// <param name="queue">投递的队列</param>
       /// <param name="IsDurability">是否需要持久化，默认为true</param>
       /// <param name="delayTime">延迟时间，单位：秒</param>
        public void SendMessage(string message, string queue, bool IsDurability = true,int delayTime=0)
        {
            context.SendConnection = RabbitMQInit.GetCurrentRabbitMQ().CreateConnection();
            using (context.SendConnection )
            {
                context.SendChannel = RabbitMQInit.GetCurrentRabbitMQ().CreateChannel(context.SendConnection,delayTime);
                 Byte deliveryMode =1;
                if (IsDurability)
                {
                    deliveryMode = 2;//持久化
                }


                using (context.SendChannel)
                {
                    if (delayTime != 0)
                    {
                        var pop = context.SendChannel.CreateBasicProperties();
                        var msg = Encoding.UTF8.GetBytes(message);
                        pop.DeliveryMode = deliveryMode;
                        pop.Expiration = (delayTime * 1000).ToString();
                        //context.SendChannel.BasicPublish(RabbitMQInit.GetCurrentRabbitMQ().config.exchanges[0].Name, "delay_" + queue, pop, msg);
                        context.SendChannel.BasicPublish(RabbitMQInit.GetCurrentRabbitMQ().config.exchanges[0].Name, "delay." + delayTime + "." + queue, pop, msg);
                    }
                    else
                    {
                        var pop = context.SendChannel.CreateBasicProperties();
                        var msg = Encoding.UTF8.GetBytes(message);
                        pop.DeliveryMode = deliveryMode;
                        context.SendChannel.BasicPublish(RabbitMQInit.GetCurrentRabbitMQ().config.exchanges[0].Name, queue, pop, msg);
                    }
                    //context.SendChannel.BasicPublish(RabbitMQInit.GetCurrentRabbitMQ().config.exchanges[0].Name, queue, pop, msg);

                }
               

            }
        }


        #region 接收消息
        public void OnListening()
        {
            try
            {
                Task.Factory.StartNew(ListenInit);
                //ListenInit();
            }
            catch (Exception)
            {
                throw;
            }
        }

        private void ListenInit()
        {

            context.ListenConnection = RabbitMQInit.GetCurrentRabbitMQ().CreateConnection();
            context.ListenConnection.ConnectionShutdown += (v, e) =>
            {
                //write logs
            };
            context.ListenChannel = RabbitMQInit.GetCurrentRabbitMQ().CreateChannel(context.ListenConnection);


            var consumer = new EventingBasicConsumer(context.ListenChannel); //事件驱动模型

            consumer.Received += Received;

            context.ListenChannel.BasicQos(0, 1, false); //每个队列一次处理一条消息，公平分发模式
            
            context.ListenChannel.BasicConsume(ListenQueueName, false, consumer);


        }

        List<AckMessage> queue = new List<AckMessage>();
        /// <summary>
        /// 接收消息处理消息
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void Received(object sender, BasicDeliverEventArgs e)
        {
            try
            {
                string message =  Encoding.UTF8.GetString(e.Body);
                
                AckMessage ackbm =null;

                //先尝试反序列化为消息确认的消息
                try
                {
                    if (message.Contains("Operation"))
                    {
                        AckMessage ack = Newtonsoft.Json.JsonConvert.DeserializeObject<AckMessage>(message);
                        Console.WriteLine(ack.MessageID);
                        Task<AckMessage> acktask = Task.Run<AckMessage>(() => { return ack; });
                        ackbm = acktask.Result;
                        queue.Add(ackbm);
                    }
                   

                }
                catch
                {

                }
                if (ackbm==null)
                {
                    Task<BaseMessage> task = Task.Run<BaseMessage>(() => { return basemessage.GetMessageResult(message); });
                    BaseMessage bm = task.Result;
                    //序列化失败或者该消息需要确认
                    if (bm == null || (bm.IsNeedAck&& queue.Where(t => t.MessageID == bm.MessageID).Count()==0))
                    {
                            //讲未能消费的消息放回队列中
                            context.ListenChannel.BasicReject(e.DeliveryTag, true);
                        
                    }
                    else
                    {
                        bool flag = false;
                        if (bm.IsNeedAck)
                        {
                            List<AckMessage> selectQueue = (from t in queue where t.MessageID==bm.MessageID orderby t.AckTime select t).ToList();
                            Console.WriteLine(selectQueue.Count);
                            foreach (var item in selectQueue)
                            {
                                flag = (item.Operation == AckOperation.确认) ? true : false;
                                queue.Remove(item);
                            }
                        }
                        if (flag)
                        {
                            if (context.ListenChannel.IsOpen)//需要注意
                            {
                                context.ListenChannel.BasicAck(e.DeliveryTag, false);
                            }
                            Task resultTask = null;
                            if (fun != null && act == null)
                            {
                                resultTask = task.ContinueWith(antecendent => { return fun(bm); });
                            }
                            else
                            {
                                resultTask = task.ContinueWith(antecendent => { act(bm); });
                            }
                            if (bm.IsNeedAck && queue.Where(t => t.MessageID == bm.MessageID).Count() > 0)
                            {

                            }
                        }
                     
                    }
                   
                    

                }

               
                
            }
            catch (Exception)
            {

                throw;
            }
        }
        #endregion

    }
}
